package com.sy.rabbitmq.qos;

import com.rabbitmq.client.*;
import com.sy.rabbitmq.TestProperties;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 消息消费限流 -消费端
 */
public class Comsumer {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 创建ConnectionFactory
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(TestProperties.getIp());
        factory.setPort(TestProperties.getPort());
        factory.setVirtualHost("/");

        //2.获取Connection
        Connection connection = factory.newConnection();

        //3.获取channel
        Channel channel = connection.createChannel();

        //4.通过channel声明一个exchange
        String exchangeName = "test_qos_exchange";
        String routeKey = "qos.#";
        String queueName = "test_qos_queue";
        channel.exchangeDeclare(exchangeName,"topic",true);

        //5.通过channel声明一个队列,绑定设置路由key
        channel.queueDeclare(queueName,true,false,false,null);
        channel.queueBind(queueName,exchangeName,routeKey);

        //6.设置qos 限流
        //prefetchSize 消息大小限制
        //prefetchCount 消息数量 在ack没有回复前多大承受消息数量
        //global 设置限流是应用到整个channel还是当前consume
        channel.basicQos(0,1,false);
        //7.限流方式必须自己手动ack不能自动ack
        channel.basicConsume(queueName,false,new MyConsumer(channel));
    }


    /**
     * 重写消费端
     */
    public static class MyConsumer extends DefaultConsumer {

        private Channel channel;
        /**
         * Constructs a new instance and records its association to the passed-in channel.
         *
         * @param channel the channel to which this consumer is attached
         */
        public MyConsumer(Channel channel) {
            super(channel);
            this.channel = channel;
        }


        /**
         * 具体消费代码编写
         * @param consumerTag
         * @param envelope
         * @param properties
         * @param body
         * @throws IOException
         */
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            System.out.println("-----------------新消费方式-----------------");
            System.out.println("consumerTag"+consumerTag);
            System.out.println("envelope"+envelope);
            System.out.println("AMQP.BasicProperties"+properties);
            System.out.println("body"+new String(body));


            //回复ack
            //deliveryTag 消息标识
            //multiple 是否批量回复
            this.channel.basicAck(envelope.getDeliveryTag(),false);
        }
    }

}
